package frontend;

import java.io.IOException;
import java.net.DatagramSocket;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.omg.CORBA.StringHolder;

import communication.Message;
import communication.ReliableUDPSender;
//import communication.ReliableUDPListener;

/**
 * This class is the implementation object for your IDL interface.
 */

public class DRSServiceServerImpl extends DRStore.DRSServicePOA {

	int firstReplicaPort;
	int secondReplicaPort;
	int thirdReplicaPort;

	String replica1, replica2, replica3;
	Sequencer ordering;

	private final ExecutorService threadPool;
	Boolean replicasAvailable = true;
	static LogFile file;
	Map <Integer, byte[]> requestBuffer;

/**
	 * Constructor for DRSServiceServerImpl 
	 * @throws SocketException 
	 */
	public DRSServiceServerImpl(String first, String second, String third) throws SocketException {
		firstReplicaPort = 61000;
		secondReplicaPort = 61001;
		thirdReplicaPort = 61002;
		replica1 = first;
		replica2 = second;
		replica3 = third;
		ordering = new Sequencer();
		threadPool = Executors.newFixedThreadPool(60);
		file = new LogFile();
		requestBuffer = new  HashMap<Integer, byte[]>();
	}

	@Override
	public void hello(StringHolder msg) {
		System.out.println("Hello "+msg.value+ "!");
		msg.value = "Welcome to DRS!";

	}

	@Override
	public String buy(String customerID, String itemID, int numberOfItem) {
		int bought;
		bought = getRService("buy", customerID, "", 0, itemID, numberOfItem);
		System.out.println("Received in buy stock: "+ bought);

		if(bought == numberOfItem)
			return customerID + ", thanks for buying " + bought +" of "+ itemID+".";
		else if (bought == -1)
			return "Sorry, server error. Please try again.";
		else
			return ("Sorry "+customerID+", low availability of "+itemID+"! Bought " + bought +".");
	}

	@Override
	public String returnStock(String customerID, String itemID, int numberOfItem) {
		int returned; 
		returned = getRService("return", customerID, "", 0, itemID, numberOfItem);
		System.out.println("Received in return stock: "+ returned);
		if(returned == 1)
			return customerID+", your request for returning "+numberOfItem+" of item "
			+itemID + " has been prosessed. Thank you!";
		else if (returned == -1)
			return "Sorry, server error. Please try again.";
		else
			return ("Sorry "+customerID+", error in processing your return request for "
					+itemID+". Please try again.");
	}

	@Override
	public String checkStock(String itemID) {
		int stock; 
		stock = getRService("checkStock", "", "", 0, itemID, 0);
		System.out.println("Received in check stock: "+ stock);
		if (stock>0){
			return stock + " of " + itemID + " is available in total.";
		}else if (stock == -1){ 
			return "Sorry, server error. Please try again.";
		} else //if stock == 0
			return itemID+ " is not available, sorry.";
	}

	@Override
	public String exchange(String customerID, String boughtItemID,
			int boughtNumber, String desiredItemID, int desiredNumber) {
		int operation; 
		operation = getRService("exchange", customerID, boughtItemID, boughtNumber, desiredItemID, desiredNumber);
		System.out.println("Received in exchange stock: "+ operation);
		if (operation==1){
			return customerID+", your exchange request was successfully processed!";
		} else if (operation == 0) {
			return "Sorry, "+customerID+", exchange is not possible: "+desiredItemID+" is not available.";
		} else if (operation == 2) {
			return "Sorry, "+customerID+", exchange is not possible: Sufficient quantity of "+boughtItemID+" is not in record.";
		} else return "Sorry, "+customerID+", server error. Please try again.";
	}

	@Override
	public void shutdown() {
		// TODO Auto-generated method stub

	}


	int getRService(String service, String customerID, String boughtItemID,
			int boughtNumber, String desiredItemID, int desiredNumber){
		byte[] request = new byte[32];
		long number = ordering.getSequenceNumber();
		int rm1Result = 0, rm2Result = 0, rm3Result = 0;

		request = packMessage(service, customerID, boughtItemID, boughtNumber, desiredItemID, desiredNumber, number);
		if (replicasAvailable) {
			try {
				System.out.println("sending message");
				try {
					sendRequest(request, number);
				} catch (TimeoutException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}

			} catch (UnknownHostException e) {
				System.out.println("Please check the HOST name and try again!!!!!!!!");
			} catch (SocketException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (ExecutionException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}

			// should get value from holdbackqueue
			rm1Result = getResultFromReplica1(number);
			rm2Result = getResultFromReplica2(number);
			rm3Result = getResultFromReplica3(number);

		}
		else{
			//hold processing by saving requests
			saveRequest(number, request);
			// not implementing reprocessing the queue
		}


		try {
			logError(rm1Result, rm2Result, rm3Result);
		} catch (UnknownHostException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		try {
			System.out.println("\nThree results: "+ rm1Result+" "+ rm2Result+" " +rm3Result);
			return getMajority(rm1Result, rm2Result, rm3Result);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (ExecutionException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return 0; 

	}

	byte[] packMessage(String type, String customerID, String boughtItemID,
			int boughtNumber, String desiredItemID, int desiredNumber, long number){
		Message packer = new Message(0);
//		System.out.println("before packing: "+ customerID+" "+desiredItemID+" "+type );
		packer.setRequestType(type);
		packer.setCustomerID(customerID);
		packer.setBoughtItemID(boughtItemID);
		packer.setBoughtNumber(boughtNumber);
		packer.setItemID(desiredItemID);
		packer.setNumberOfItems(desiredNumber);
		packer.setSequencerID(number);
		byte[] packed =packer.generateMessage();
//		String received = new String(packed, Charset.defaultCharset());
//		System.out.println("packed: "+received);
		return packed;
	}

	void sendRequest(byte[] request, long number) throws UnknownHostException, SocketException, InterruptedException, ExecutionException, TimeoutException{
		// open socket, send packed request, receive packed reply

		Future<Integer> receivedFromReplica1 = null, receivedFromReplica2=null, receivedFromReplica3 = null ;
		// sending to RM1
		int fCount=0; int fAck = 0;
		do{
			ReliableUDPSender r1ClientUDP = new ReliableUDPSender(new DatagramSocket(), replica1, firstReplicaPort);
			r1ClientUDP.sendMessage = request;
			System.out.println("Submitted "+ number+ " to rm1." + new String (request, Charset.defaultCharset()));

			receivedFromReplica1 = threadPool.submit(r1ClientUDP);
			fCount++;
			try {
				fAck = receivedFromReplica1.get(3, TimeUnit.SECONDS);
			} catch (TimeoutException e) {
				System.out.println("Request timedout for "+ number);
			}
			System.out.println("Ack received from RM1 for sqID "+ number+" is "+ fAck);
		}while (fAck!= number && fCount<3);



		// sending to RM2
		int sCount=0; int sAck = 0;
		do{
			ReliableUDPSender r2ClientUDP = new ReliableUDPSender(new DatagramSocket(), replica2, secondReplicaPort);
			r2ClientUDP.sendMessage = request;
			System.out.println("Submitted "+ number+ " to rm2." + new String (request, Charset.defaultCharset()));

			receivedFromReplica2 = threadPool.submit(r2ClientUDP);
			sCount++;
			try {
				sAck = receivedFromReplica2.get(3, TimeUnit.SECONDS);
			} catch (TimeoutException e) {
				System.out.println("Request timedout for "+ number);
			}
			System.out.println("Ack received from RM2 for sqID "+ number+" is "+ sAck);
		}while (sAck!= number && sCount<3);

		 

		// sending to RM3
		int tCount=0; int tAck = 0;
		do{
			ReliableUDPSender r3ClientUDP = new ReliableUDPSender(new DatagramSocket(), replica3, thirdReplicaPort);
			r3ClientUDP.sendMessage = request;
			System.out.println("Submitted "+ number+ " to rm3." + new String (request, Charset.defaultCharset()));

			receivedFromReplica3 = threadPool.submit(r3ClientUDP);
			tCount++;
			try {
				tAck = receivedFromReplica3.get(3, TimeUnit.SECONDS);
			} catch (TimeoutException e) {
				System.out.println("Request timedout for "+ number);
			}
			System.out.println("Ack received from RM3 for sqID "+ number+" is "+ tAck);
		}while (tAck!= number && tCount<3);
	}

	void saveRequest(long number, byte[] request){
		int key = (int)number;
		requestBuffer.put(key, request);
	}

	int getResultFromReplica1(long number){
		int tCount=0; int got = -9;
		byte[] tosend = new byte[16];
		tosend = packMessage("reply", "", "", 0, "", 0, number);
		do{
			ReliableUDPSender r1 = null;
			try {
				r1 = new ReliableUDPSender(new DatagramSocket(), replica1, firstReplicaPort);
			} catch (UnknownHostException e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			} catch (SocketException e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			}
			r1.sendMessage = tosend;
			System.out.println("Requesting reply for "+ number+ " at rm1." + new String (tosend, Charset.defaultCharset()));

			Future<Integer>  answer = threadPool.submit(r1);
			tCount++;

			try {
				got = answer.get(8, TimeUnit.SECONDS);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (ExecutionException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (TimeoutException e) {
				System.out.println("Request timedout for "+ number);
			}
			System.out.println("Answer received from RM1 for sqID "+ number+" is "+ got);
		}while (got==999999 && tCount<3);
		return got;
	}


	int getResultFromReplica2(long number){
		int tCount=0; int got = -9;
		byte[] tosend = new byte[16];
		tosend = packMessage("reply", "", "", 0, "", 0, number);
		do{
			ReliableUDPSender r2 = null;
			try {
				r2 = new ReliableUDPSender(new DatagramSocket(), replica2, secondReplicaPort);
			} catch (UnknownHostException e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			} catch (SocketException e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			}
			r2.sendMessage = tosend;
			System.out.println("Requesting reply for "+ number+ " at rm2." + new String (tosend, Charset.defaultCharset()));

			Future<Integer>  answer = threadPool.submit(r2);
			tCount++;

				try {
					got = answer.get(8, TimeUnit.SECONDS);
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (ExecutionException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (TimeoutException e) {
				System.out.println("Request timedout for "+ number);
			}
			System.out.println("Answer received from RM2 for sqID "+ number+" is "+ got);
		}while (got==999999 && tCount<3);
		return got;
	}

	int getResultFromReplica3(long number){
		int tCount=0; int got = -9;
		byte[] tosend = new byte[16];
		tosend = packMessage("reply", "", "", 0, "", 0, number);
		do{
			ReliableUDPSender r3 = null;
			try {
				r3 = new ReliableUDPSender(new DatagramSocket(), replica3, thirdReplicaPort);
			} catch (UnknownHostException e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			} catch (SocketException e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			}
			r3.sendMessage = tosend;
			System.out.println("Requesting reply for "+ number+ " at rm3." + new String (tosend, Charset.defaultCharset()));

			Future<Integer>  answer = threadPool.submit(r3);
			tCount++;

			try {
				got = answer.get(8, TimeUnit.SECONDS);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (ExecutionException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (TimeoutException e) {
				System.out.println("Request timedout for "+ number);
			}
			System.out.println("Answer received from RM3 for sqID "+ number+" is "+ got);
		}while (got==999999 && tCount<5);
		return got;
	}

	int getMajority(int first, int second, int third) throws InterruptedException, ExecutionException{
		if(first == second){
			return first;
		} else if(first == third){
			return third;
		} else if(second == third){
			return second;
		} else 
			return -1; //when no majority
	}

	// should implement separately
	void logError(int one, int two, int three) throws UnknownHostException {

		if(one==two){
			if(two==three){
				//nothing to log
			} else {
				// third is wrong
				System.out.println("Error found in replica 3.");
				file.updateFile(0, 0, 1);
			}
		} else if(two==three) {
			//first is wrong
			System.out.println("Error found in replica 1.");
			file.updateFile(1, 0, 0);
		} else if(one==three){
			//second is wrong
			System.out.println("Error found in replica 2.");
			file.updateFile(0, 1, 0);
		} else {
			//nothing matches
		}
	}


	class LogFile{
		int [] rmFile;
		ReliableUDPSender toRM;
		DatagramSocket socket = null;
		String servername;
		int portNo = 0;

		public LogFile() throws SocketException{
			rmFile = new int [4];
			socket = new DatagramSocket();
			portNo = 60500;
			servername = "localhost"; // change if run in a different host
		}

		void updateFile(int one, int two, int three) throws UnknownHostException{
			rmFile[1] += one;
			rmFile[2] += two;
			rmFile[3] += three;
			checkFile();
		}

		void checkFile() throws UnknownHostException{
			for (int i=1;i<4;i++){
				if(rmFile[i] == 3){
					try {
						sendError(i);
					} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}
			}
		}

		void sendError(int rmNumber) throws IOException{

			Future<Integer> fromRM;
			int count=0; int ack = 0;
			do{
				toRM = new ReliableUDPSender(socket, servername, portNo);
				printFile();
				// make replicasAvailable to false
				replicasAvailable = false;
				toRM.sendMessage = Integer.toString(rmNumber).getBytes();
				System.out.println("Sending faulty replica number "+rmNumber+" to Replica Manger...");

				fromRM = threadPool.submit(toRM);
				count++;
				try {
					ack = fromRM.get(5, TimeUnit.SECONDS);
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (ExecutionException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (TimeoutException e) {
					System.out.println("Reset request for replica "+rmNumber+" timedout.");
				}
				System.out.println("Ack received from RM after resetting "+ rmNumber+" is "+ ack);
			}while (ack!= 1 && count<3);

			clearError(rmNumber);
			// make replicasAvailable to true
			replicasAvailable = true;
		}

		void clearError(int index) {
			rmFile[index] = 0;

		}

		void printFile(){
			System.out.println("\n\nRm Error file info:");
			for (int i=1;i<4;i++){
				System.out.println("RM"+i+":  "+rmFile[i]);
			}
		}
	}

}
